#include <errno.h>
#include "server.h"

bool ExistConn( Connection*pConn,long long seq )
{ 
	return  pConn->lSeq == seq; 
}

int SendNow( HandleBuffer*buffer )
{
	if ( !ExistConn(buffer->pConn,buffer->lSeq) )
	{
		return CONN_NOT_EXIST;
	}
	int iSend = 0;
	int fd = buffer->pConn->fd;
	while ( true )//发送一个buffer
	{
		iSend = send( fd,&buffer->packet+buffer->iPos,buffer->iSize-buffer->iPos,0 );
		if ( iSend <0 )
		{
			if ( errno == EAGAIN )//
			{
				perror("send buffer is full\n");
				return SOCKET_EAGAIN;
			}
			else if( errno == EINTR )
			{
				continue;
			}
			else{
				perror( "socket error\n" );
				return SOCKET_ERROR;
			}
		}
		else if( iSend == 0  )
		{
			perror("socket is closed\n");
			return SOCKET_CLOSED;
		}
		else{

			buffer->iPos+=iSend;
			if ( buffer->iPos == buffer->iSize )//完成
			{
				return 0;
			}
		}
	}
	return 0;
}

int SendQueue(std::list<HandleBuffer*>*pSendQueue)
{
	if ( pSendQueue->size() )//发送队列不为空
	{
		int iSend = 0;
		int r = 0;
		HandleBuffer*buffer = NULL;
		do //先将队列数据发送完
		{
			buffer = pSendQueue->front();
			int fd = buffer->pConn->fd;
			if ( buffer->iPos >= buffer->iSize )//
			{
				printf("error:buffer pos large than size\n");
				return ERROR_BUFFER_SIZE;
			}
			r = SendNow( buffer );
			if ( r!=0 )
			{
				return r;
			}
			pSendQueue->pop_front();
			delete buffer;
		}while(pSendQueue->size());

	}
	return 0;
}

CLogicThread::CLogicThread(CServer*p)
{
	m_pIO = p;
}

void CLogicThread::Run()
{ 
	m_pIO->LogicLoop();
}


CServer::CServer( char*sIP,const short iPort ):
CEpollSvrIO(sIP,iPort)
{
	m_pTimeOutHeap = new CMinHeap<CTimeEvent>(10);
	m_pConnMgr = new ConnectionMgr( MAX_CONN_COUNT );
	m_bExit = false;
	m_pLogicThread = new CLogicThread(this);
	m_pLogicThread->Start();
}
int CServer::OnClose( Connection*pConn )
{

}

int CServer::OnConnect( int connfd,sockaddr*addr )
{
	static long long lSeq = 1;
	Connection *pConn = m_pConnMgr->GetConnection();
	if ( !pConn )
	{
		printf( "not enough connection buf\n" );
		//CloseConnection()
		close(connfd);
		return NOT_ENOUGH_CONNBUF;
	}
	SetNonBlocking(connfd);
	pConn->fd = connfd;
	pConn->lSeq = lSeq++;
	pConn->iConnectTime = time(NULL);
	//memcpy(&pConn->sockAddr,addr,sizeof(sockaddr));
	sockaddr_in*p = (sockaddr_in*)addr;
	memcpy(&pConn->addr,&p->sin_addr,sizeof(in_addr));
	AddConn(pConn,READ_EVENT|WRITE_EVENT|CLEAR_EVENT);
}

//尽量读取
int CServer::OnRead( Connection*pConn )
{	
	//printf("OnRead.....\n");
	int r=0;
	while ( 0==r )
	{
		r = ReadOnePack(pConn);
	}
	return r;

}

int CServer::OnWrite( Connection*pConn )
{
	//Connection*pConn = (Connection*)pEv->events.ptr;
	m_SendLock.Lock();
	int r = SendQueue(pConn->pSendQueue);
	m_SendLock.Unlock();
	return r;
}

int  CServer::CloseConnection( Connection*pConn )
{
	//printf("socket close\n");
	OnClose(pConn);
	pConn->lSeq = 0;
	std::list<HandleBuffer*>::iterator itr;
	close( pConn->fd );
	m_pConnMgr->FreeConnection(pConn);
}


int  CServer::ReadOnePack(Connection*pConn )
{
	int fd = pConn->fd;
	int pos = pConn->iReadPos;
	int iHeadLen = GetPackHeaderLen();
	int needRead = 0;
	Packet*packet=NULL;
	int iRet = 0;
	if( pos< iHeadLen )//还没有读取完头部
	{
		needRead = iHeadLen - pos;
		iRet = Read( pConn,needRead);
		if ( iRet == SOCKET_CLOSED||iRet == SOCKET_ERROR)
		{
			CloseConnection( pConn );
			return READ_ONE_PACK_NOT_FINISHED;
		}
		if ( iRet == SOCKET_EAGAIN )
		{
			return READ_ONE_PACK_NOT_FINISHED;
		}
		
		packet = (Packet*)pConn->sReadBuf;
		if ( iHeadLen + packet->len >MAX_PACKET_LEN )
		{
			CloseConnection( pConn );
			return ERROR_PACKET_LEN;
		}
		//读包体
		iRet = Read( pConn,packet->len);
		if (  iRet == SOCKET_CLOSED||iRet == SOCKET_ERROR )
		{
			CloseConnection( pConn );
			return READ_ONE_PACK_NOT_FINISHED;
		}
		if ( iRet == SOCKET_EAGAIN )
		{
			return READ_ONE_PACK_NOT_FINISHED;
		}
		EnQueue(m_RecvQueue,pConn);

	}
	else{
		packet = (Packet*)pConn->sReadBuf;
		needRead = packet->len - pos-iHeadLen;
		iRet= Read( pConn,needRead);
		if (  iRet == SOCKET_CLOSED||iRet == SOCKET_ERROR )
		{
			CloseConnection( pConn );
			return -1;
		}
		if ( iRet == SOCKET_EAGAIN )
		{
			return READ_ONE_PACK_NOT_FINISHED;
		}

		EnQueue(m_RecvQueue,pConn);
	}
	pConn->iReadPos = 0;
	return 0;//
}

int  CServer::EnQueue(const std::list<HandleBuffer*>&queue,Connection*pConn)
{
	Packet*p=(Packet*)pConn->sReadBuf;
	int iSize = sizeof(Connection*)+sizeof(long long)+2*sizeof(int)+sizeof(Packet)+p->len;//待处理的节点buf大小
	HandleBuffer*pBuffer = (HandleBuffer*)malloc( iSize);
	pBuffer->pConn = pConn;
	pBuffer->lSeq = pConn->lSeq;
	pBuffer->iPos = 0;
	pBuffer->iSize = sizeof(Packet)+p->len;//包头+包体
	memcpy( &pBuffer->packet,p,sizeof(Packet)+p->len );
	m_RecvLock.Lock();
	m_RecvQueue.push_back( pBuffer );
	m_RecvLock.Unlock();
	m_HaveWorkConn.Signal();
}

//逻辑线程函数
int CServer::LogicLoop()
{

	while(!m_bExit)
	{
		ProcessTimeOut();
		m_RecvLock.Lock();//加锁可能需要一些时间
		while ( m_RecvQueue.size()==0 )
		{
			int timeout = ProcessTimeOut();//返回毫秒数
			printf("ProcessTimeOut 返回:%d\n",timeout);
			if ( timeout>0 )
			{
				struct timespec stAbsTime;
				TimevalDelay(timeout,&stAbsTime);
				m_HaveWorkConn.TimedWait(m_RecvLock,&stAbsTime);
			}
			else
				m_HaveWorkConn.Wait(m_RecvLock);
			
		}
		HandleBuffer*buff = m_RecvQueue.front();
		m_RecvQueue.pop_front();
		m_RecvLock.Unlock();
		Handle( buff );
	}

}

int CServer::Send( HandleBuffer*result )
{
	if ( !ExistConn(result->pConn,result->lSeq))//连接已经不存在
	{
		return CONN_NOT_EXIST;
	}
	int iSend = 0;
	int r = 0;
	Connection*pConn = result->pConn;
	m_SendLock.Lock();
	if ( pConn->pSendQueue->size() )//发送队列不为空,就加入队列
	{
		printf( "send queue is not empty\n" );
		pConn->pSendQueue->push_back(result);
		//r = SendQueue(pConn->pSendQueue);
		m_SendLock.Unlock();
		return SEND_DELAYED;
	}
	else//队列为空，直接发送
	{
		//printf( "send queue is empty,send at once\n" );
		if( (r=SendNow(result))!=0)//没有发送完毕
		{
			printf("not all data is send\n");
			pConn->pSendQueue->push_back(result);
			m_SendLock.Unlock();
			return SEND_DELAYED;
		}
		//printf("all data is send\n");
		delete result;//发送完毕需要删除
		m_SendLock.Unlock();
		return 0;
	}
	return 0;
}

int CServer::Handle( HandleBuffer*buf )
{
	
}

int CServer::Loop()
{
	while(!m_bExit)
	{
		ProcessEvent();
	}
}

int CServer::ProcessTimeOut()
{
	int timeout = -1;
	int now  =GetRunningTime();
	printf("now time is %d\n",now);
	while ( true )
	{
		CTimeEvent*event = m_pTimeOutHeap->GetMin();
		if (event)
		{
			printf("timer:%d,time out is %d\n",event->GetID(),event->GetTimeOut());
			if( now>=event->GetTimeOut()) //超时
			{
				m_pTimeOutHeap->ExtractMin();
				OnTimer(event);
				delete event;
			}
			else
			{
				timeout = event->GetTimeOut()-now;
				return timeout;
			}
		}
		else 
			return timeout;
	}
	return -1;
}

//在IO线程中
int CServer::OnTimer(CTimeEvent*pEvent)
{
	printf("Timer:%d is time out\n",pEvent->GetID());
}

void CServer::Exit()
{
	m_bExit = true;
}

int	CServer::DelayCall( int timeout,cbTimeOut cbfunc, void*arg)
{

	//printf("now is %d,DelayCall:%d\n",GetRunningTime(),timeout );
	CTimeEvent*pEvent = new CTimeEvent(timeout,cbfunc,arg);
	m_pTimeOutHeap->Insert(pEvent);
}
